package com.aotain.hbase;


import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Increment;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.FilterList.Operator;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
import org.apache.hadoop.hbase.util.Bytes;



public class HbaseCommon {
	static Configuration cfg = null;

	private static Connection connection = null;

	private static String _zooServer = "192.168.5.95";
//	private static String _zooServer = "119.29.88.56,119.29.151.41,119.29.181.248";
//	private static String _zooServer = "132.126.32.4";

	private static void InitConnection()
	{
		Configuration HBASE_CONFIG = HBaseConfiguration.create();

		//		ZkKafkaConfig zkConfig = new ZkKafkaConfig("/home/storm/config/dbconfig.ini");
		//		if(zkConfig.getZkConnServer()!=null && !zkConfig.getZkConnServer().isEmpty())
		//			_zooServer = zkConfig.getZkConnServer();

		HBASE_CONFIG.set("hbase.zookeeper.quorum", _zooServer);
		//HBASE_CONFIG.set("hbase.zookeeper.quorum", "shanghai-cm-5");  // 
		HBASE_CONFIG.set("hbase.zookeeper.property.clientPort","2181");
		//HBASE_CONFIG.set("fs.defaultFS","hdfs://hive-2:8020");
		//HBASE_CONFIG.set("zookeeper.znode.parent","/user/hbase");
		//0.96  
		//System.setProperty("hadoop.home.dir", "C:\\cygwin\\usr\\hadoop");
		cfg = HBaseConfiguration.create(HBASE_CONFIG);
		try {
			connection = ConnectionFactory.createConnection(HBASE_CONFIG);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		System.out.println("IP:"+cfg.get("hbase.zookeeper.quorum"));
		System.out.println("PORT:"+cfg.get("hbase.zookeeper.property.clientPort"));
	}


	
	public static void incr(String tableName,String rowKey,String col,long value) {
		InitConnection();
		Increment increment = new Increment(Bytes.toBytes(rowKey));
		TableName TABLE_NAME = TableName.valueOf(tableName);
		try {
			Table table = connection.getTable(TABLE_NAME);
			String cf = col.split(":",-1)[0];
      	  	String column = col.split(":",-1)[1];
			increment.addColumn(Bytes.toBytes(cf), Bytes.toBytes(column), 1);
			table.increment(increment);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	/**
	 * 根据时间戳范围查找数据
	 * @param tablename
	 * @param StartStamptime
	 * @param EndStamptime
	 * @throws Exception
	 */
	public static void getAllData(String tablename,Date StartStamptime,Date EndStamptime) throws Exception{


		InitConnection();

		TableName TABLE_NAME = TableName.valueOf(tablename);
		Table table = connection.getTable(TABLE_NAME);
		Scan s = new Scan();
		s.setTimeRange(StartStamptime.getTime(), EndStamptime.getTime());
		ResultScanner ss = table.getScanner(s);
		int count = 0;
		for (Result r : ss)
		{//
			String rowkey = Bytes.toString(r.getRow());
			String row = "";

			row = rowkey + "/";
			for(Cell c : r.listCells()){
				String ff = Bytes.toString(CellUtil.cloneFamily(c));//
				String cc = Bytes.toString(CellUtil.cloneQualifier(c));//

				if(!tablename.equals("SDS_ABNORMAL_LOG") && (cc.contains("LOW") || cc.contains("MIDDLE")
						|| cc.equals("PV")))
				{
					Long bb = Bytes.toLong(CellUtil.cloneValue(c));//
					//System.out.print(ff + ":" + cc + "=" + bb + "/");
					row = row + ff + ":" + cc + "=" + bb + "/";
				}
				else
				{
					String bb = Bytes.toString(CellUtil.cloneValue(c));
					//System.out.print(ff + ":" + cc + "=" + bb + "/");
					row = row + ff + ":" + cc + "=" + bb + "/";
				}


			}

			//if(rownum <= limit)
			{
				System.out.print(row);
				System.out.println();
			}

			count++;
			if(count >=  1000)
				break;
		}
	}


	public static void AddData(String tablename,String rowkey,String family,String column,String value)
	{


		try {
			HTable table = new HTable(cfg,tablename);
			Put p = new Put(rowkey.getBytes());
			p.add(family.getBytes(),column.getBytes(),value.getBytes("UTF8"));
			table.put(p);
		}  catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}


	/**
	 * 
	 * @param tablename
	 * @param startkey
	 * @param endkey
	 * @throws Exception
	 */
	public static void getData(String tablename,String startkey,String endkey,int limit) throws Exception
	{

		InitConnection();

		TableName TABLE_NAME = TableName.valueOf(tablename);
		Table table = connection.getTable(TABLE_NAME);

		//HTable table = new HTable(cfg,tablename);
		int rownum = 0;
		try {

			Scan sc = new Scan(startkey.getBytes(), endkey.getBytes());
			sc.setReversed(true);
			sc.setCaching(10);

			Filter filter = new PageFilter(limit);


			sc.setFilter(filter);  
			sc.setMaxVersions(1);
			ResultScanner ss = table.getScanner(sc);
			System.out.println("start===showOneRecordByRowKey==row: "+"\n");

			for (Result r : ss)
			{//
				String rowkey = Bytes.toString(r.getRow());
				String row = "";
				rownum++;
				row = rowkey + "/";
				for(Cell c : r.listCells()){
					String ff = Bytes.toString(CellUtil.cloneFamily(c));//
					String cc = Bytes.toString(CellUtil.cloneQualifier(c));//

					if(!tablename.equals("SDS_ABNORMAL_LOG") && (cc.contains("LOW") || cc.contains("MIDDLE")
							|| cc.equals("PV")|| cc.equals("STPROVINCEPV") || cc.equals("STREAMOCTETS") || cc.equals("STREAMPACKET")||cc.equals("HTTPNUM") ) )
					{
						Long bb = Bytes.toLong(CellUtil.cloneValue(c));//
						//System.out.print(ff + ":" + cc + "=" + bb + "/");
						row = row + ff + ":" + cc + "=" + bb + "/";
					}
					else
					{
						String bb = Bytes.toString(CellUtil.cloneValue(c));
						//System.out.print(ff + ":" + cc + "=" + bb + "/");
						row = row + ff + ":" + cc + "=" + bb + "/";
					}

				}

				//if(rownum <= limit)
				{
					System.out.print(row);
					System.out.println();
				}
				//else
				{
					//   break;
				}
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		finally
		{
			table.close();
			connection.close();
		}
		System.out.println("end===========row:" + rownum);
	}


	/**
	 * 
	 * @param tablename
	 * @param startkey
	 * @param endkey
	 * @throws Exception
	 */
	public static void getDataStatByTime(String tablename,String startkey,
			String endkey,String columnName) throws Exception
	{

		InitConnection();

		TableName TABLE_NAME = TableName.valueOf(tablename);
		Table table = connection.getTable(TABLE_NAME);

		HashMap<String,Integer> map = new HashMap<String,Integer>();
		//HTable table = new HTable(cfg,tablename);
		int rownum = 0;
		try {

			Scan sc = new Scan(startkey.getBytes(), endkey.getBytes());
			//sc.setReversed(true);
			//sc.setBatch(100);
			//sc.setCaching(10);
			//sc.setCaching(100);

			//		   Filter filter = new PageFilter(limit);
			//		   sc.setFilter(filter);  

			//sc.setMaxVersions(1);
			ResultScanner ss = table.getScanner(sc);
			System.out.println("start===getDataStatByTime==row: "+"\n");

			for (Result r : ss)
			{//

				rownum++;
				Cell c = r.getColumnLatestCell("cf".getBytes(), columnName.getBytes());

				if(c == null)
					continue;
				String sTime = Bytes.toString(CellUtil.cloneValue(c));
				String key = sTime.substring(0,12);
				if(map.containsKey(key))
				{
					int count = map.get(key);
					count = count + 1;
					map.put(key, count);
				}
				else
				{
					map.put(key, 1);
				}
			}


			for(String key : map.keySet())
			{
				String row = String.format("%s [%d]", key,map.get(key));
				System.out.println(row);
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		finally
		{
			table.close();
			connection.close();
		}
		System.out.println("end===========row:" + rownum);
	}


	public static void getDataByColumn(String tablename,
			String startkey,String endkey,String column,int limit) throws Exception
	{

		InitConnection();

		TableName TABLE_NAME = TableName.valueOf(tablename);
		Table table = connection.getTable(TABLE_NAME);

		//HTable table = new HTable(cfg,tablename);
		int rownum = 0;
		try {

			Scan sc = new Scan(startkey.getBytes(), endkey.getBytes());
			//		   sc.setReversed(true);
			//		   sc.setCaching(10);
			Filter filter = new PageFilter(limit);
			sc.setFilter(filter);  

			ResultScanner ss = table.getScanner(sc);
			System.out.println("start===showOneRecordByRowKey==row: "+"\n");

			for (Result r : ss)
			{//
				Cell cell = r.getColumnLatestCell(Bytes.toBytes("cf"), 
						Bytes.toBytes(column));

				if(cell != null)
					System.out.println(Bytes.toString(CellUtil.cloneValue(cell)));
			}

		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		finally
		{
			table.close();
			connection.close();
		}
		System.out.println("end===========row:" + rownum);
	}


	public static void QueryByCondition2(String tablename,
			String startRow,String endRow,
			String[] columns,
			String column1,String value1) { 

		InitConnection();

		Table table = null;
		try { 

			TableName TABLE_NAME = TableName.valueOf(tablename);
			table = connection.getTable(TABLE_NAME);

			Scan scan = new Scan(); 
			scan.setReversed(false);
			scan.setStartRow(startRow.getBytes());
			scan.setStopRow(endRow.getBytes());


			//List<Filter> filters = new ArrayList<Filter>();
//			PageFilter page = new PageFilter(100l);
			//	    		filters.add(new PageFilter(100l));
			//	    		filters.add(new SingleColumnValueFilter(Bytes.toBytes("cf"), Bytes.toBytes(column1), CompareOp.NOT_EQUAL, Bytes.toBytes(value1)));
			FilterList filterList = new FilterList(Operator.MUST_PASS_ONE);
			SingleColumnValueFilter colfilter1 = new SingleColumnValueFilter(Bytes.toBytes("cf"), Bytes.toBytes(column1), CompareOp.EQUAL, Bytes.toBytes("10"));
			SingleColumnValueFilter colfilter2 = new SingleColumnValueFilter(Bytes.toBytes("cf"), Bytes.toBytes(column1), CompareOp.EQUAL, Bytes.toBytes("4"));

			filterList.addFilter(colfilter2);
			filterList.addFilter(colfilter1);
			//filterList.addFilter(page);






			scan.setFilter(filterList);
			//scan.setFilter(page);

			System.out.println(filterList.toString());

			for(String col : columns)
			{
				scan.addColumn("cf".getBytes(), col.getBytes());
			}


			//	            Filter filter = new SingleColumnValueFilter(Bytes.toBytes(column1), null, CompareOp.EQUAL, Bytes 
			//	                    .toBytes(value1)); //
			int rownum = 0;
			ResultScanner rs = table.getScanner(scan);

			//for (Result r : rs) {


			for (Result r : rs) {
				String rowkey = Bytes.toString(r.getRow());
				String row = "";
				row = rowkey + "/";
				for(Cell c : r.listCells()){
					String ff = Bytes.toString(CellUtil.cloneFamily(c));//
					String cc = Bytes.toString(CellUtil.cloneQualifier(c));//

					if(!tablename.equals("SDS_IDC_LOG") && (cc.contains("LOW") || cc.contains("MIDDLE")
							|| cc.equals("PV")))
					{
						Long bb = Bytes.toLong(CellUtil.cloneValue(c));//
						//System.out.print(ff + ":" + cc + "=" + bb + "/");
						row = row + ff + ":" + cc + "=" + bb + "/";
					}
					else
					{
						String bb = Bytes.toString(CellUtil.cloneValue(c));
						//System.out.print(ff + ":" + cc + "=" + bb + "/");
						row = row + ff + ":" + cc + "=" + bb + "/";
					}


				}

				if(rownum == 100)
				{
					System.out.print("ROW--100");
					System.out.println();

				}
				else
				{
					System.out.print(row);
					System.out.println();
				}
				rownum++;

			} 

			System.out.println("end===========row:" + rownum);
		} catch (Exception e) { 

			e.printStackTrace(); 

		} 
		finally
		{
			try {
				table.close();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	} 
	
	/**
	 * 过滤字段查询
	 * @param tablename
	 * @param startRow
	 * @param endRow
	 * @param column1
	 * @param value1
	 */
	public static void QueryByConditionByOneValue(String tablename,
			String startRow,String endRow,String column1,String value1) { 

		InitConnection();

		Table table = null;
		try { 

			TableName TABLE_NAME = TableName.valueOf(tablename);
			table = connection.getTable(TABLE_NAME);

			Scan scan = new Scan(); 
			scan.setReversed(false);
			scan.setStartRow(startRow.getBytes());
			scan.setStopRow(endRow.getBytes());

			FilterList filterList = new FilterList(Operator.MUST_PASS_ALL);
			SingleColumnValueFilter colfilter1 
				= new SingleColumnValueFilter(Bytes.toBytes("cf"),
						Bytes.toBytes(column1), 
						CompareOp.EQUAL, Bytes.toBytes(value1));
			filterList.addFilter(colfilter1);
			scan.setFilter(filterList);
		
			System.out.println(filterList.toString());
			int rownum = 0;
			ResultScanner rs = table.getScanner(scan);
			for (Result r : rs) {
				String rowkey = Bytes.toString(r.getRow());
				String row = "";
				row = rowkey + "/";
				for(Cell c : r.listCells()){
					String ff = Bytes.toString(CellUtil.cloneFamily(c));//
					String cc = Bytes.toString(CellUtil.cloneQualifier(c));//

					if(!tablename.equals("SDS_IDC_LOG") && (cc.contains("LOW") 
							|| cc.contains("MIDDLE") || cc.equals("PV") 
							|| cc.equals("STREAMOCTETS")|| cc.equals("STREAMPACKET")
							|| cc.equals("HTTPNUM")))
					{
						Long bb = Bytes.toLong(CellUtil.cloneValue(c));//
						//System.out.print(ff + ":" + cc + "=" + bb + "/");
						row = row + ff + ":" + cc + "=" + bb + "/";
					}
					else
					{
						String bb = Bytes.toString(CellUtil.cloneValue(c));
						//System.out.print(ff + ":" + cc + "=" + bb + "/");
						row = row + ff + ":" + cc + "=" + bb + "/";
					}
				}

				System.out.print(row);
				System.out.println();

			} 
		} catch (Exception e) { 

			e.printStackTrace(); 

		} 
		finally
		{
			try {
				table.close();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	} 

	public static void QueryByCondition3(String tablename,
			String column1,String value1,
			String column2,String value2,
			String column3,String value3) { 

		try { 

			HTable table = new HTable(cfg,tablename);

			List<Filter> filters = new ArrayList<Filter>(); 

			Filter filter1 = new SingleColumnValueFilter(Bytes 

					.toBytes(column1), null, CompareOp.EQUAL, Bytes 

					.toBytes(value1)); 

			filters.add(filter1); 

			Filter filter2 = new SingleColumnValueFilter(Bytes 

					.toBytes(column2), null, CompareOp.EQUAL, Bytes 

					.toBytes(value2)); 

			filters.add(filter2); 



			Filter filter3 = new SingleColumnValueFilter(Bytes 

					.toBytes(column3), null, CompareOp.EQUAL, Bytes 

					.toBytes(value3)); 

			filters.add(filter3); 



			FilterList filterList1 = new FilterList(filters); 



			Scan scan = new Scan(); 

			scan.setFilter(filterList1); 

			ResultScanner rs = table.getScanner(scan); 

			for (Result rr : rs) { 

				System.out.println("rowkey: " + new String(rr.getRow())); 

				for (Cell c : rr.listCells()) { 

					System.out.println("Qualifier "+ Bytes.toString(CellUtil.cloneQualifier(c))

							+"====Value:" + Bytes.toString(CellUtil.cloneValue(c))); 
				} 

			} 

			rs.close(); 



		} catch (Exception e) { 

			e.printStackTrace(); 

		} 



	} 

	public static void main(String[] args){
		try{
			//HbaseCommon.getDataByColumn("SDS_IDC_STAT_D", "20151103", "20151105", "ATTSITENUM",200);
			//			
			//			ZooKeeper z = new ZooKeeper("hadoop-r720-1",3000, null);
			//			 zk = new ZooKeeper(hostPort, 3000, this);
			//			 DataMonitor  dm = new DataMonitor(zk, znode, null, this);
			//		        
			//			List<String> list = z.getChildren("/project/apollo/dbserver", false);
			//			
			//			String findIp = "202.156.98.66";
			//			String IP1 = findIp.substring(0,findIp.lastIndexOf("."));
			//			System.out.println(IP1);
			File workaround = new File("."); 
			System.getProperties().put("hadoop.home.dir", workaround.getAbsolutePath()); 
			new File("./bin").mkdirs();  
//			HbaseCommon.incr("t3", "cheng", "cf:col", 0);
			while(true)
			{


				BufferedReader strin=new BufferedReader(new InputStreamReader(System.in));
				String welcome = "1:按TABLENAME/STARTROW/ENDROW 查询所有列(倒序  前100条) \r\n";
				welcome = welcome + "2:统计日志每分钟的记录数(倒序)  \r\n";
				welcome = welcome + "3:字段条件过滤(ABRNORMAL = 3 100)  \r\n";
				welcome = welcome + "4:根据时间戳范围查找数据(TOP 100)  \r\n";
				welcome = welcome + "5:自定义条件过滤  \r\n";
				welcome = welcome + "请选择:";

				System.out.print(welcome);  

				String type = strin.readLine();  

				if(type.equals("1"))
				{
					System.out.println("TABLENAME:");  
					String tablename = strin.readLine();  

					System.out.println("STARTROW:");  
					String startrow = strin.readLine();  

					System.out.println("ENDROW:");  
					String endrow = strin.readLine();
					HbaseCommon.getData(tablename, startrow, endrow,1000);

				}
				else if(type.equals("2"))
				{
					System.out.println("TABLENAME:");  
					String tablename = strin.readLine();  

					System.out.println("STARTROW:");  
					String startrow = strin.readLine();  

					System.out.println("ENDROW:");  
					String endrow = strin.readLine();
					HbaseCommon.getDataStatByTime(tablename, startrow, endrow, "ACCESSTIME");
				}
				else if(type.equals("3"))
				{
					System.out.println("TABLENAME:");  
					String tablename = strin.readLine();  

					System.out.println("STARTROW:");  
					String startrow = strin.readLine();  

					System.out.println("ENDROW:");  
					String endrow = strin.readLine();
					String[] cols = new String[] {"DESTIP","ACCESSTIME","ABRNORMAL","DESC","ATTNUM","PROVINCE"};
					HbaseCommon.QueryByCondition2(tablename, startrow, endrow,cols ,"ABRNORMAL","2");
				}
				else if(type.equals("4"))
				{
					System.out.println("TABLENAME:");  
					String tablename = strin.readLine();  

					System.out.println("STARTDATE:");  
					String startrow = strin.readLine();  

					System.out.println("ENDDATE:");  
					String endrow = strin.readLine();
					//String[] cols = new String[] {"DESTIP","ACCESSTIME","DESC","ATTNUM","EVALUATE"};

					SimpleDateFormat df1 = new SimpleDateFormat("yyyyMMdd");
					Date StartStamptime = df1.parse(startrow);
					Date EndStamptime = df1.parse(endrow);
					HbaseCommon.getAllData(tablename, StartStamptime, EndStamptime);
					//		            HbaseCommon.QueryByCondition2(tablename, startrow, endrow ,"EVALUATE","100");
				}
				else if(type.equals("5"))
				{
					System.out.println("TABLENAME:");  
					String tablename = strin.readLine();  

					System.out.println("STARTDATE:");  
					String startrow = strin.readLine();  

					System.out.println("ENDDATE:");  
					String endrow = strin.readLine();
					
					System.out.println("COLUMN:");  
					String column = strin.readLine();
					
					System.out.println("VALUE:");  
					String value = strin.readLine();
					
					HbaseCommon.QueryByConditionByOneValue(tablename, startrow, 
							endrow,column, value);
				}
				
				

				System.out.println("OK");
				System.out.println("Continue? Y/N");
				String YesOrNo = strin.readLine();

				if(YesOrNo.equals("Y") || YesOrNo.equals("y") || YesOrNo.equals("Yes") 
						|| YesOrNo.equals("yes") || YesOrNo.equals("YES"))
				{
					continue;
				}
				else
				{
					return;
				}

			}
		}
		catch(Exception e)
		{
			e.printStackTrace();
		}
	}

	public static List<String> QueryByKey(String tablename,String startRow,String endRow,String[] columns) { 
		List<String> list = new ArrayList<String>();
		InitConnection();

		Table table = null;
		try { 

			TableName TABLE_NAME = TableName.valueOf(tablename);
			table = connection.getTable(TABLE_NAME);

			Scan scan = new Scan(); 
			//scan.setReversed(true);
			scan.setStartRow(startRow.getBytes());
			scan.setStopRow(endRow.getBytes());

			for(String col : columns){
				scan.addColumn("cf".getBytes(), col.getBytes());
			}

			ResultScanner rs = table.getScanner(scan); 
			for (Result r : rs) {
				String rowkey = Bytes.toString(r.getRow());
				String row = "";
				row = rowkey;
				for(Cell c : r.listCells()){
					String ff = Bytes.toString(CellUtil.cloneFamily(c));
					String cc = Bytes.toString(CellUtil.cloneQualifier(c));
					if(cc.contains("SESSIONMIDDLE") || cc.contains("PORTLOW") || cc.contains("PV"))
					{
						// String bb = Bytes.toString(CellUtil.cloneValue(c));
						Long bb = Bytes.toLong(CellUtil.cloneValue(c));
						row = row + "_" + bb;
					}
				}
				list.add(row);
			}
		} catch (Exception e) { 
			e.printStackTrace(); 
		} 
		finally
		{
			try {
				table.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return list;
	} 



	public static List<String> QueryByKey2(String tablename,String startRow,String endRow,String[] columns) { 
		List<String> list = new ArrayList<String>();
		InitConnection();

		Table table = null;
		try { 

			TableName TABLE_NAME = TableName.valueOf(tablename);
			table = connection.getTable(TABLE_NAME);

			Scan scan = new Scan(); 
			//scan.setReversed(true);
			scan.setStartRow(startRow.getBytes());
			scan.setStopRow(endRow.getBytes());

			for(String col : columns){
				scan.addColumn("cf".getBytes(), col.getBytes());
			}

			ResultScanner rs = table.getScanner(scan); 
			for (Result r : rs) {
				String rowkey = Bytes.toString(r.getRow());
				String row = "";
				row = rowkey;
				Long bb=0L;

				for(String ent :columns){
					Cell cell = r.getColumnLatestCell("cf".getBytes(), ent.getBytes());
					if(cell == null){
						bb=0L;
					}else{
						bb = Bytes.toLong(CellUtil.cloneValue(cell));
					}
					row = row + "_" + bb;
				} 
				list.add(row);
			}
		} catch (Exception e) { 
			e.printStackTrace(); 
		} 
		finally
		{
			try {
				table.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return list;
	} 


}
